package org.egomsl.mw.table;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.regex.Pattern;

public class TableDesc {
    public static String partitionDirPattern = "part-\\d+-\\d+";
    public static String partitionDirFormat = "part-%04d-%04d";

    private Path tablePath;

    private int numVirtualNode;

    private int[] partions;

    private List<Path> partitionDirs;

    public TableDesc() {
        tablePath = null;
        numVirtualNode = 0;
        partions = null;
        partitionDirs = null;
    }

    public TableDesc(Path tablePath, int numVirtualNode, int[] partions, List<Path> partitionDirs) {
        this.tablePath = tablePath;
        this.numVirtualNode = numVirtualNode;
        this.partions = partions;
        this.partitionDirs = partitionDirs;
    }

    /**
     *
     * @param tablePath
     * @param numVirtualNode
     * @param numPartion
     */
    public TableDesc(Path tablePath, int numVirtualNode, int numPartion) {
        this.tablePath = tablePath;
        this.numVirtualNode = numVirtualNode;

        partions = new int[numPartion];
        int step = numVirtualNode / numPartion;
        for(int i=0;i<numPartion;i++) {
            partions[i] = step * i;
        }

        partitionDirs = new ArrayList<>();
        for(int i=0;i<numPartion-1;i++) {
            partitionDirs.add(new Path(tablePath,
                    String.format(partitionDirFormat, partions[i], partions[i+1]))
            );
        }
        partitionDirs.add(new Path(tablePath,
                String.format(partitionDirFormat, partions[numPartion-1], numVirtualNode))
        );
    }

    public Path getTablePath() {
        return tablePath;
    }

    public void setTablePath(Path tablePath) {
        this.tablePath = tablePath;
    }

    public int getNumVirtualNode() {
        return numVirtualNode;
    }

    public void setNumVirtualNode(int numVirtualNode) {
        this.numVirtualNode = numVirtualNode;
    }

    public int[] getPartions() {
        return partions;
    }

    public void setPartions(int[] partions) {
        this.partions = partions;
    }

    public List<Path> getPartitionDirs() {
        return partitionDirs;
    }

    public void setPartitionDirs(List<Path> partitionDirs) {
        this.partitionDirs = partitionDirs;
    }


    public static TableDesc loadFromFs(FileSystem fs, Path tableDir) throws IOException {

        FileStatus[] fileStatuses = fs.listStatus(tableDir,
                path->Pattern.matches(partitionDirPattern, path.getName()));

        if(fileStatuses == null) return null;

        Arrays.sort(fileStatuses, Comparator.comparing(FileStatus::getPath));

        int numVirtualNode = 0;
        int[] partions = new int[fileStatuses.length];
        List<Path> partitionDirs = new ArrayList<>(partions.length);

        int lastEndHash = 0;
        for(int i=0;i<fileStatuses.length;i++) {
            FileStatus status = fileStatuses[i];
            String[] fields = status.getPath().getName().split("-");
            int startHash = Integer.valueOf(fields[1]);
            int endHash = Integer.valueOf(fields[2]);

            if(startHash != lastEndHash) {
                throw  new RuntimeException("Error: there's a hole in table partition from "
                        + lastEndHash + " to " + startHash);
            }

            partions[i] = startHash;
            partitionDirs.add(status.getPath());

            lastEndHash = endHash;
        }
        numVirtualNode = lastEndHash;

        return new TableDesc(tableDir, numVirtualNode, partions, partitionDirs);
    }
}
